/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package com.creditease.shield.flow.client;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import com.creditease.shield.flow.common.msg.Request;
import com.creditease.shield.flow.common.msg.Response;
import com.creditease.shield.flow.handler.FlowNoClientHandler;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * Modification of {@link EchoClient} which utilizes Java object serialization.
 */
public final class NettyClient implements IClient{
	private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);
    
	@Value("${client.worker.group.threads}")
	private int workerGroupThreads;
    
    public Channel channel;
    
    private EventLoopGroup workerGroup;
    
    private Bootstrap clientBootStrap;
    
    private volatile boolean closed;
    
    private FlowNoClientHandler flowNoClientHandler = new FlowNoClientHandler();
    
    public Response syncSend(Request request) throws InterruptedException {
        System.out.println("send request:"+request);
        channel.writeAndFlush(request).sync();
        return flowNoClientHandler.send(request,null);
    }

    public Response asyncSend(Request request,Pair<Long,TimeUnit> timeout) throws InterruptedException {
        channel.writeAndFlush(request);
        return flowNoClientHandler.send(request,timeout);
    }

	@Override
	public InetSocketAddress getRemoteAddress() {
		SocketAddress remoteAddress = channel.remoteAddress();
        if (!(remoteAddress instanceof InetSocketAddress)) {
            throw new RuntimeException("Get remote address error, should be InetSocketAddress");
        }
        return (InetSocketAddress) remoteAddress;
	}

	
	 public boolean isClosed() {
	        return closed;
	    }

	@Override
	public void close() {
		logger.info("destroy client resources");
        if (null == channel) {
            logger.error("channel is null");
        }
        closed = true;
        workerGroup.shutdownGracefully();
        channel.closeFuture().syncUninterruptibly();
        workerGroup = null;
        channel = null;
		
	}

	public void connect(final InetSocketAddress socketAddress) {
        try{
        	//配置客户端线程组
        	workerGroup = new NioEventLoopGroup(workerGroupThreads);
        	//配置客户端启动辅助类  
            clientBootStrap = new Bootstrap();
            clientBootStrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline()
                                    //处理失败重连
                                    .addFirst(new ChannelInboundHandlerAdapter() {
                                        @Override
                                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                            super.channelInactive(ctx);
                                            ctx.channel().eventLoop().schedule(new Runnable() {
                                                @Override
                                                public void run() {
                                                    doConnect(socketAddress);
                                                }
                                            },1, TimeUnit.SECONDS);
                                        }
                                    })
                                    //每隔30s的时间触发一次userEventTriggered的方法，并且指定IdleState的状态位是WRITER_IDLE
                                    .addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS))
                                    .addLast(new ConnectorIdleStateTrigger())
                                    .addLast(new ObjectEncoder())
                                    .addLast(new ObjectDecoder(1024*1024,ClassResolvers.cacheDisabled(this.getClass().getClassLoader())))
                                    .addLast(flowNoClientHandler);
                        }
                    });
            doConnect(socketAddress);
        }catch (Exception e){
            logger.error(e.getMessage(), e);
        }
    }
    
    
    private void doConnect(final InetSocketAddress socketAddress) {
        logger.info("trying to connect server:{}",socketAddress);
        if (closed) {
            return;
        }

        ChannelFuture future = clientBootStrap.connect(socketAddress);
        future.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture f) throws Exception {
                if (f.isSuccess()) {
                    logger.info("connected to {}", socketAddress);
                } else {
                    logger.info("connected to {} failed",socketAddress);
                    f.channel().eventLoop().schedule(new Runnable() {
                        @Override
                        public void run() {
                            doConnect(socketAddress);
                        }
                    }, 1, TimeUnit.SECONDS);
                }
            }
        });

        channel = future.syncUninterruptibly().channel();
    }
    
}
